/* *
 * This file is part of Feng
 *
 * Copyright (C) 2009 by LScube team <team@lscube.org>
 * See AUTHORS for more details
 *
 * feng is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * feng is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with feng; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 *
 * */

#include "bufferqueue.h"
#include <stdio.h>
#include <string.h>
#include <unistd.h>

typedef struct {
    double foo;
    double bar;
    size_t data_size;
    guchar data[];
} Stuff;

static int awake = 0;
static int stop_fill = 0;

static GMutex *mux;

static void fill_cb(gpointer cons_p, gpointer prod_p)
{
    BufferQueue_Consumer *consumer = cons_p;
    BufferQueue_Producer *producer = prod_p;
    Stuff *buffer = g_malloc0(sizeof(Stuff) + 2000);

    g_mutex_lock(mux);

    while (bq_consumer_unseen(consumer) < 16) {
        buffer->foo++;
        buffer->bar--;
        buffer->data_size = 2000;
        memset(buffer->data, 'a', 14);
        bq_producer_put(producer, g_memdup (buffer, sizeof(Stuff) + 2000));
        if(awake++ > 11) {
            sleep(1);
            fprintf(stderr, "Sleeping %p\n", consumer);
            awake = 0;
        }
        if (stop_fill) goto exit;
    }
    exit:
    g_free(buffer);
    g_mutex_unlock(mux);
}


void test_2()
{
    int size = 10, i;
    int count = 24, j;
    Stuff *ret;
    Stuff *buffer = g_malloc0(sizeof(Stuff) + 2000);
    BufferQueue_Consumer *cons[size];
    BufferQueue_Producer *prod = bq_producer_new();
    GThreadPool *pool = g_thread_pool_new(fill_cb, prod, 6, FALSE, NULL);
    mux = g_mutex_new();

//init

    buffer->foo = 0;
    buffer->bar = 10;
    buffer->data_size = 2000;
    buffer->data[0] = 'a';

    bq_producer_put(prod, g_memdup (buffer, sizeof(Stuff) + 2000));
//consume stuff

    for(j = 0; j <count; j++) {
        BufferQueue_Consumer *c = bq_consumer_new(prod);
        fprintf(stderr, "cc %p: new consumer\n", c);
        ret = bq_consumer_get(c);
        if (ret)
            fprintf(stderr, "cc %p: %p %f %f \n",c, ret, ret->bar, ret->foo);
        else
            fprintf(stderr, "cc %p: NULL \n", c);
        fprintf(stderr, "cc %p: consumer_free\n", c);
        // g_thread_pool_push (pool, cons[i], NULL);
        bq_consumer_free(c);
    }

    for (i = 0; i < size; i++) {
        fprintf(stderr,"---- Allocating consumer %d ", i);
        cons[i] = bq_consumer_new(prod);
        fprintf(stderr,"%p\n", cons[i]);
    }

    for (j=0; j < count; j++) {
        for (i = 0; i < size; i++) {
            ret = bq_consumer_get(cons[i]);
            if (ret)
                fprintf(stderr, "Foo: %d %p %f %f\n",
                        i, ret, ret->bar, ret->foo);
            else
                fprintf(stderr, "Foo: %d NULL\n", i);

            if(!bq_consumer_move(cons[i]))
                fprintf(stderr, "  %d no next\n", i);
            else
                fprintf(stderr, "  %d next %p\n",
                        i, bq_consumer_get(cons[i]));
            g_thread_pool_push (pool, cons[i], NULL);
        }
    }

    fprintf(stderr, "---- Killing _put thread\n");
    stop_fill = 1;
    g_thread_pool_free(pool, TRUE, TRUE);
    for (i = 0; i < size/2; i++)
        bq_consumer_get(cons[i]);
    for (i = 0; i < size; i++)
        if (cons[i]) {
            fprintf(stderr, "---- Deallocating consumer %d %p\n",
                          i, cons[i]);
            bq_consumer_free(cons[i]);
        }
    bq_producer_unref(prod);
    g_mutex_free(mux);
    g_free(buffer);
}
